package com.qit.prototype.flink.performance;

import java.util.List;
import java.util.Map;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.functions.PatternProcessFunction;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

public class TestCase2 {
	
	public TestCase2() throws Exception {
		
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		env.setStreamTimeCharacteristic(TimeCharacteristic.ProcessingTime);
		
		DataStreamSource<String> dataStreamSource = env
				.socketTextStream("localhost", 11002, "\0");
		
		DataStream<Tuple2<String, Integer>> ds = dataStreamSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {

			private static final long serialVersionUID = 8084559234308500699L;

			@Override
			public void flatMap(String arg0,
					Collector<Tuple2<String, Integer>> arg1) throws Exception {
				for (String word: arg0.split("\\|")) {
	                arg1.collect(new Tuple2<String, Integer>(word, 1));
	            }
			}
			
		});
		
		
		Pattern<Tuple2<String, Integer>, ?> pattern = Pattern.<Tuple2<String, Integer>>begin("start").where(new SimpleCondition<Tuple2<String, Integer>>() {

			private static final long serialVersionUID = -4529582378775261658L;

			@Override
			public boolean filter(Tuple2<String, Integer> tick) throws Exception {
//				return true;
				if ("100000".equals(tick.f0)) {
					return true;
				}
				return false;
			}
			
			
		});
		
		PatternStream<Tuple2<String, Integer>> patternStream = CEP.pattern(ds, pattern);
		
		patternStream.process(new PatternProcessFunction<Tuple2<String, Integer>, Tuple2<String, Integer>>() {

			private static final long serialVersionUID = -2740368404996075800L;

			@Override
			public void processMatch(
					Map<String, List<Tuple2<String, Integer>>> arg0,
					org.apache.flink.cep.functions.PatternProcessFunction.Context arg1,
					Collector<Tuple2<String, Integer>> arg2) throws Exception {
				System.out.println(System.currentTimeMillis());
				System.out.println(arg0);
			}
			
		});
		
        env.execute("Performance");
		
	}

}
